package net.wlm.realtime.funcation;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.wlm.realtime.utils.HbaseUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import java.util.Collections;
import java.util.concurrent.*;
import java.util.function.Supplier;
/*
 * 异步请求Hbase,从维表获取数据
 */
public class RequestDicDimAsyncFunction extends RichAsyncFunction<String, String> {
    /**
     * 线程池操作对象
     */
    private ExecutorService executorService;
    @Override
    public void open(Configuration parameters) throws Exception {
        // a-初始化线程池
        executorService =  Executors.newScheduledThreadPool(10);
    }
    @Override
    public void asyncInvoke(String s, ResultFuture<String> resultFuture) throws Exception {
        // 解析JSON字符串
        JSONObject jsonObject = JSON.parseObject(s);
        String rowKey = jsonObject.getString("source_type");
        // 异步请求HBAse获取数据
        Future<String> result = executorService.submit(
                new Callable<String>() {
                    @Override
                    public String call() throws Exception {
                        return HbaseUtil.getData("dim_base_dic", rowKey, "info", "dic_name");
                    }
                }
        );
        // 异步返回结果
        CompletableFuture.supplyAsync(new Supplier<String>() {
            @Override
            public String get() {
                try {
                    String requestValue = result.get();
                    // 添加字段值
                    jsonObject.put("source_type_name",requestValue);
                } catch (ExecutionException  | InterruptedException e) {
                    // 设置字段为：exception
                    jsonObject.put("source_type_name","exception");
                }
                // 返回
                return jsonObject.toJSONString();
            }
        }).thenAccept((String dbResult) -> {
            resultFuture.complete(Collections.singleton(dbResult));
        });
    }
    @Override
    public void timeout(String input, ResultFuture<String> resultFuture) throws Exception {
        // 解析JSON
        JSONObject jsonObject = JSON.parseObject(input);
        // 设置字段为：timeout
        jsonObject.put("source_type_name","timeout");
        // 返回
        resultFuture.complete(Collections.singleton(jsonObject.toJSONString()));
    }
    @Override
    public void close() throws Exception {
        super.close();
    }
}